package stateful;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Map state 使用
 * 统计每个省分城市的销售金额
 * 状态分为
 * keyed state ValueStatue MapState ListState
 * non-key state 又称为 OperatorState(ListState BroadcastState)
 */
public class MapStateDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 开启checkpoint
        env.enableCheckpointing(5000);

        DataStreamSource<String> lines = env.socketTextStream("hadoop1", 8888);

        lines.map(new MapFunction<String, Tuple3<String, String, Double>>() {
            @Override
            public Tuple3<String, String, Double> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(fields[0], fields[1], Double.parseDouble(fields[2]));
            }
        }).keyBy(t -> t.f0).process(new KeyedProcessFunction<String, Tuple3<String, String, Double>, Tuple3<String, String, Double>>() {

            // 定义状态
            private transient MapState<String, Double> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 初始化时恢复状态
                // 1.定义一个状态描述器
                MapStateDescriptor<String, Double> mapStateDescriptor = new MapStateDescriptor<String, Double>("map_state", String.class, Double.class);
                // 2.从运行时上下文中获取状态数据
                mapState = getRuntimeContext().getMapState(mapStateDescriptor);
            }

            @Override
            public void processElement(Tuple3<String, String, Double> value, Context ctx, Collector<Tuple3<String, String, Double>> out) throws Exception {
                String city = value.f1;
                Double historyMoney = mapState.get(city);
                if (historyMoney == null) {
                    historyMoney = 0.0;
                }
                Double totalMoney = historyMoney + value.f2;
                mapState.put(city, totalMoney);
                value.f2 = totalMoney;
                out.collect(value);
            }
        }).print();

        env.execute("");
    }
}
